package com.allen.flink.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * 功能: 单词计数
 *
 * @date: 2020-03-19 23:18
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class WordCount {
    
    public static void main(String[] args) {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //获取输入对象
        DataSet<String> text = env.fromElements("Hooray! It's snowing!", " It's time to make a snowman.", "James runs out.", " He makes a big pile of snow");
        
        //计数
        
        DataSet<Tuple2<String, Integer>> counts = text.flatMap(new LineSplitter()).groupBy(0) //key 分组
                                                      .sum(1); //value sum
        
        //打印
        try {
            counts.print();
        } catch (Exception e) {
            e.printStackTrace();
        }
        
    }
    
    //分隔
    private static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(final String word, final Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] tokens = word.toLowerCase().split("\\W+");//非单词分隔
            
            for (final String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}

